package org.chen.service;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApi;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.chen.model.AccessLogMessage;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
@RocketMQMessageListener(
    topic = "article_access_log",
    consumerGroup = "article_access_log_group"
)
public class AccessLogConsumer implements RocketMQListener<AccessLogMessage> {

    private final StringRedisTemplate redisTemplate;
    private final WriteApi writeApi;
    private final static String COUNT_KEY = "article:count:";

    public AccessLogConsumer(InfluxDBClient influxDBClient,
                             StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.writeApi = influxDBClient.makeWriteApi();
    }

    @Override
    public void onMessage(AccessLogMessage message) {
        log.info("Process access log message success:{}", message);
        try {
            // 1. 更新Redis计数器
            updateRedisCounter(message);

            // 2. 写入时序数据库(InfluxDB)
            saveToInfluxDB(message);

        } catch (Exception e) {
            log.error("Process access log message failed", e);
        }
    }

    private void updateRedisCounter(AccessLogMessage message) {
        String articleId = message.getArticleId();
        String ip = message.getIp();

        updateViewCount(articleId, ip);
    }

    /**
     * 更新访问计数
     */
    private void updateViewCount(String articleId, String ip) {
        try {
            String viewKey = "article:view:" + articleId + ":" + ip;
            // 检查5分钟内是否已计数
            if (Boolean.TRUE.equals(redisTemplate.opsForValue()
                    .setIfAbsent(viewKey, "1", 5, TimeUnit.MINUTES))) {

                CompletableFuture.runAsync(() -> {
                    // 先增加Redis计数，定期同步到DB
                    String countKey = COUNT_KEY + articleId;
                    redisTemplate.opsForValue().increment(countKey);

                    // PV计数器
                    String pvKey = "article:pv:" + articleId;
                    redisTemplate.opsForValue().increment(pvKey);

                    // UV统计(使用HyperLogLog)
                    String uvKey = "article:uv:" + articleId;
                    redisTemplate.opsForHyperLogLog().add(uvKey, ip);

                    // 当天PV统计
                    String dailyKey = "article:pv:" + articleId + ":" +
                            LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE);
                    redisTemplate.opsForValue().increment(dailyKey);

                    // 当天UV统计
                    String dailyUVKey = "article:uv:" + articleId + ":" +
                            LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE);
                    redisTemplate.opsForValue().increment(dailyUVKey);
                });
            }
        } catch (Exception e) {
            log.error("Update view count failed", e);
        }
    }

    private void saveToInfluxDB(AccessLogMessage message) {
        // 构建InfluxDB数据点
        Point point = Point.measurement("article_views")
            .addTag("article_id", message.getArticleId())
            .addTag("ip", message.getIp())
            .addField("duration", message.getDuration())
            .addField("user_agent", message.getUserAgent())
            .addField("referer", message.getReferer())
            .time(message.getTimestamp(), WritePrecision.MS);

        // 异步写入InfluxDB
        writeApi.writePoint(point);
    }
}
